package com.gxlevi.task

import com.gxlevi.bean.ClickLogWide
import com.gxlevi.util.HBaseUtil
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.sink.SinkFunction

case class ChannelFreshness(
                             var channelID: String,
                             var date: String,
                             var newCount: Long,
                             var oldCount: Long
                           )

object ChannelFreshnessTask extends BaseTask[ChannelFreshness] {
  override def map(clickWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelFreshness] = {
    clickWideDataStream.flatMap {
      clickLog =>
        List(
          ChannelFreshness(clickLog.channelID, clickLog.yearMonthDayHour, clickLog.isNew, isOld(clickLog.isNew, clickLog.isHourNew)),
          ChannelFreshness(clickLog.channelID, clickLog.yearMonthDay, clickLog.isNew, isOld(clickLog.isNew, clickLog.isDayNew)),
          ChannelFreshness(clickLog.channelID, clickLog.yearMonth, clickLog.isNew, isOld(clickLog.isNew, clickLog.isMonthNew))
        )
    }
  }

  override def keyBy(mapStream: DataStream[ChannelFreshness]): KeyedStream[ChannelFreshness, String] = {
    mapStream.keyBy {
      freshness =>
        freshness.channelID + freshness.date
    }
  }

  override def reduce(windowedStream: WindowedStream[ChannelFreshness, String, TimeWindow]): DataStream[ChannelFreshness] = {
    val reduce = windowedStream.reduce {
      (t1, t2) =>
        ChannelFreshness(t1.channelID, t1.date, t1.newCount + t2.newCount, t1.oldCount + t2.oldCount)
    }
    reduce.print()
    reduce
  }

  override def sinkToHBase(reduceDataStream: DataStream[ChannelFreshness]): Unit = {
    reduceDataStream.addSink(new SinkFunction[ChannelFreshness] {
      override def invoke(value: ChannelFreshness): Unit = {
        tableName = "channel_fresh"
        val newCountColName = "newCount"
        val oldCountColName = "oldCount"
        rowKey = value.channelID + ":" + value.date

        val map: Map[String, String] = HBaseUtil.getMapData(tableName, rowKey, clfName, List(newCountColName, oldCountColName))
        val totalNewCount = getTotal(map, newCountColName, value.newCount)
        val totalOldCount = getTotal(map, oldCountColName, value.oldCount)

        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          channelIdColumn -> value.channelID,
          dateColumn -> value.date,
          newCountColName -> totalNewCount,
          oldCountColName -> totalOldCount
        ))
      }
    })
  }
}
